// Database datapump implementation
// Copyright 2006 (C) Ralph Thomas

#include <datapumps/database.h>
#include <binder.hpp>
#include <util/db.h>
#include <util/resources.h>
#include <adobe/array.hpp>
#include <boost/bind.hpp>
#include <ext/hash_fun.h>

#include <iostream>

namespace bfs = boost::filesystem;
namespace sgi = __gnu_cxx;

namespace datapumps {
	struct db_row_result_t {
		adobe::dictionary_t row_m;
		std::size_t hash_m;

		db_row_result_t() {}
		db_row_result_t( const adobe::dictionary_t& row ) : row_m( row ), hash_m( 0 ) {}
		db_row_result_t( const db_row_result_t& rhs ) : row_m( rhs.row_m ), hash_m( rhs.hash_m ) {}
		std::size_t hash() const {
			if( hash_m != 0 ) return hash_m;
			if( row_m.empty() ) return 0;
			adobe::dictionary_t::const_iterator first( row_m.begin() );
			adobe::dictionary_t::const_iterator last( row_m.end() );
			sgi::hash<const char*> hasher;
			std::size_t hash = 0;
			while( first != last ) {
				//
				// The values are always strings.
				//
				std::string value = first->second.get<std::string>();
				hash ^= hasher( value.c_str() );
				first++;
			}
			const_cast<db_row_result_t*>( this )->hash_m = hash;
			return hash_m;
		}
		bool operator==( const db_row_result_t& rhs ) const {
			//
			// Check identity.
			//
			if( this == &rhs ) return true;
			//
			// Check dictionary size.
			//
			if( row_m.size() != rhs.row_m.size() ) return false;
			//
			// Check hash. Expensive the first time.
			//
			if( hash() != rhs.hash() ) return false;
			//
			// Check dictionary. Expensive!
			//
			return row_m == rhs.row_m;
		}
	};
	
	static void add_row( adobe::array_t& target, const adobe::dictionary_t& source, const adobe::dictionary_t::value_type& extra ) {
		adobe::dictionary_t mod_source( source );
		mod_source.insert( extra );
		target.push_back( adobe::value_t( mod_source ) );//db_row_result_t( mod_source ) ) );
	}
	class db_query_t : public adobe::binder_t {
		util::db_ptr_t	db_m;		///< Our database.
		std::string	db_name_m;	///< Name of our database.
		std::string	query_m;	///< Query into database.
		adobe::binder_helper_t query_pump_m;	///< Query pump.
		adobe::binder_helper_t db_pump_m;	///< Db name pump.
		adobe::sheet_t::monitor_value_t value_monitor_m;
		adobe::sheet_t::monitor_active_t active_monitor_m;
		adobe::sheet_t&	sheet_m;
		bfs::path	scope_path_m;
		adobe::value_t	original_db_pump;
		adobe::value_t	original_query;
		boost::signals::scoped_connection db_connection_m;
		//
		/// Call our query and publish results. We are never active because
		/// we are always read-only.
		//
		void update() {
			if( !db_m || query_m.empty() ) {
				if( value_monitor_m ) value_monitor_m( adobe::value_t() );
				if( active_monitor_m ) active_monitor_m( false );
				return;
			}
			if( value_monitor_m ) {
				adobe::array_t result_set;
				adobe::dictionary_t::value_type db_name_extra;
				db_name_extra.first = adobe::static_name_t( ".db_name" );
				db_name_extra.second = adobe::value_t( db_name_m );
				try {
					db_m->query( query_m, boost::bind( &add_row, boost::ref( result_set ), _1, db_name_extra ) );
				} catch( ... ) {
					value_monitor_m( adobe::value_t() );
					if( active_monitor_m ) active_monitor_m( false );
					return;
				}
				value_monitor_m( adobe::value_t( result_set ) );
			}
			if( active_monitor_m ) active_monitor_m( false );
		}
		void update_query( const adobe::value_t& query ) {
			std::string new_query = query.get<std::string>();
			if( query_m == new_query ) return;
			query_m = new_query;
			update();
		}
		void update_db( const adobe::value_t& db_value ) {
			std::string db_name = db_value.get<std::string>();
			if( db_name == db_name_m ) return;
			db_name_m = db_name;
			db_m = util::db::fromPath( scope_path_m / bfs::path( db_name_m, bfs::native ) );
			//
			// Make sure that we get notifications from the database.
			//
			db_connection_m = db_m->on_change( boost::bind( &db_query_t::update, this ) );
			update();
		}
	  public:
		db_query_t( const bfs::path& scope_path, const adobe::value_t& db_name, const adobe::value_t& query, adobe::sheet_t& sheet ) : query_pump_m( sheet ), db_pump_m( sheet ), sheet_m( sheet ), scope_path_m( scope_path ), original_db_pump( db_name ), original_query( query ) {
			query_pump_m.initialize( query, boost::bind( &db_query_t::update_query, this, _1 ), 0 );
			db_pump_m.initialize( db_name, boost::bind( &db_query_t::update_db, this, _1 ), 0 );
		}
		virtual ~db_query_t() {}
		virtual void set( const adobe::value_t& val ) {}
		virtual void set_signals( const adobe::sheet_t::monitor_value_t& monitor_value, const adobe::sheet_t::monitor_active_t& monitor_active ) {
			value_monitor_m = monitor_value;
			active_monitor_m = monitor_active;
			update();
		}
		virtual void substitute( const adobe::name_t& name, const adobe::value_t& val ) {
			query_pump_m.substitute( name, val );
			db_pump_m.substitute( name, val );
			update();
		}
		virtual void invoke() {}
		virtual adobe::binder_ptr_t copy() const {
			return adobe::binder_ptr_t( new db_query_t( scope_path_m, original_db_pump, original_query, sheet_m ) );
		}
	};
	adobe::value_t make_db_query_datapump( const bfs::path& scope_path, const adobe::dictionary_t& parameters, adobe::sheet_t& sheet ) {
		adobe::value_t db_name( parameters.at( adobe::static_name_t( "db" ) ) );
		adobe::value_t query( parameters.at( adobe::static_name_t( "query" ) ) );
		return adobe::value_t( adobe::binder_ptr_t( new db_query_t( scope_path, db_name, query, sheet ) ) );
	}
	//
	/// Used by db_row_t and db_value_t to generate the UPDATE statement by
	/// substituting new values in. Substitutes one value from the given
	/// dictionary into the string, returns the result. If not substitutions
	/// can be made (none remain in the given string) then the same string
	/// is returned.
	///
	/// \param	s	the string to substitute into.
	/// \param	d	the dictionary of strings to substitute from.
	/// \return	the substituted string, or s if no substitutions could be made.
	//
	static std::string substitute_dict( const std::string& s, const adobe::dictionary_t& d ) {
		std::string::size_type at = s.find( '@' );
		std::string::size_type semi = s.find( ';', at );
		if( at == std::string::npos ) return s;
		if( semi == std::string::npos ) return s;
		std::string token = s.substr( at + 1, semi - at - 1 );
		std::string replacement;
		adobe::name_t name( token.c_str() );
		if( d.count( name ) ) replacement = d.get<std::string>( name );
		std::string new_s( s );
		new_s.replace( at, semi - at + 1, replacement );
		return new_s;
	}
	class db_row_t : public adobe::binder_t {
		db_row_result_t		row_m;
		adobe::name_t		column_m;
		std::string		update_m;
		adobe::binder_helper_t	row_pump_m;
		adobe::sheet_t::monitor_active_t	active_monitor_m;
		adobe::sheet_t::monitor_value_t		value_monitor_m;
		bfs::path		scope_path_m;
		//
		// for making copies.
		// 
		adobe::sheet_t&		sheet_m;
		adobe::value_t		row_value_m;
		//
		/// Update the value of the row.
		///
		/// \param	val	the new row value.
		//
		void update_row( const adobe::value_t& val ) {
			db_row_result_t row( val.get<adobe::dictionary_t>() );//db_row_result_t>() );
			if( row == row_m ) return;
			row_m = row;
			update();
		}
		//
		/// Update the value we're reflecting into the
		/// pipeline.
		//
		void update() {
			if( row_m.row_m.count( column_m ) ) {
				if( active_monitor_m ) active_monitor_m( !update_m.empty() );
				if( value_monitor_m ) value_monitor_m( row_m.row_m.at( column_m ) );
				return;
			}
			if( active_monitor_m ) active_monitor_m( false );
			if( value_monitor_m ) value_monitor_m( adobe::value_t() );
		}
	  public:
		db_row_t( const bfs::path scope_path, const adobe::value_t& row, const adobe::name_t& column, const std::string& update, adobe::sheet_t& sheet )
		 : column_m( column ), update_m( update ), row_pump_m( sheet ), scope_path_m( scope_path ), sheet_m( sheet ), row_value_m( row ) {
			row_pump_m.initialize( row, boost::bind( &db_row_t::update_row, this, _1 ), 0 );
		}
		virtual ~db_row_t() {}
		virtual void set( const adobe::value_t& val ) {
			//
			// Walk through the update string, looking for "@" and ";".
			// 
			std::string update( update_m );
			std::string new_update( update_m );
			adobe::dictionary_t subst_dict( row_m.row_m );
			subst_dict.set( adobe::static_name_t( "NEWVAL" ), adobe::value_t( util::db::escape( val.get<std::string>() ) ) );
			do {
				update = new_update;
				new_update = substitute_dict( update, subst_dict );
			} while( update != new_update );
			//
			// The row has to have a ".db_name" member.
			//
			std::string db_name = row_m.row_m.get<std::string>( adobe::static_name_t( ".db_name" ) );
			util::db_ptr_t db = util::db::fromPath( scope_path_m / bfs::path( db_name, bfs::native ) );
			try {
				db->execute( update );
			} catch( std::exception& ex ) {
				//
				// XXX: remove debug code.
				//
				std::cerr << "ERROR: " << ex.what() << " RUNNING: " << update << std::endl;
			}
		}
		virtual void set_signals( const adobe::sheet_t::monitor_value_t& monitor_value, const adobe::sheet_t::monitor_active_t& monitor_active ) {
			value_monitor_m = monitor_value;
			active_monitor_m = monitor_active;
			update();
		}
		virtual void substitute( const adobe::name_t& name, const adobe::value_t& val ) {
			row_pump_m.substitute( name, val );
		}
		virtual void invoke() {}
		virtual adobe::binder_ptr_t copy() const {
			return adobe::binder_ptr_t( new db_row_t( scope_path_m, row_value_m, column_m, update_m, sheet_m ) );
		}
	};
	adobe::value_t make_db_row_datapump( const bfs::path& scope_path, const adobe::dictionary_t& parameters, adobe::sheet_t& sheet ) {
		adobe::value_t row( parameters.at( adobe::static_name_t( "row" ) ) );
		adobe::name_t column( parameters.get<std::string>( adobe::static_name_t( "column" ) ).c_str() );
		std::string update;
		if( parameters.count( adobe::static_name_t( "update" ) ) )
			update = parameters.get<std::string>( adobe::static_name_t( "update" ) );
		return adobe::value_t( adobe::binder_ptr_t( new db_row_t( scope_path, row, column, update, sheet ) ) );
	}
	//
	/// The db_value datapump. Extracts a single value from the database
	/// and is able to update it.
	//
	class db_value_t : public adobe::binder_t {
		std::string				update_m;
		std::string				query_m;
		std::string				db_name_m;
		util::db_ptr_t				db_m;
		adobe::binder_helper_t			query_pump_m;
		adobe::binder_helper_t			db_pump_m;
		adobe::binder_helper_t			update_pump_m;
		adobe::sheet_t::monitor_active_t	active_monitor_m;
		adobe::sheet_t::monitor_value_t		value_monitor_m;
		boost::signals::scoped_connection	db_connection_m;
		bfs::path				scope_path_m;
		//
		// for making copies
		//
		adobe::sheet_t&				sheet_m;
		adobe::value_t				db_value_m;
		adobe::value_t				query_value_m;
		adobe::value_t				update_value_m;

		static void getResult( adobe::value_t& result, const adobe::dictionary_t& row ) {
			if( result != adobe::value_t() ) return;
			if( !row.size() ) return;
			result = row.begin()->second;
		}
		void update() {
			if( !db_m || query_m.empty() ) {
				if( active_monitor_m )	active_monitor_m( false );
				if( value_monitor_m )	value_monitor_m( adobe::value_t() );
				return;
			}
			if( value_monitor_m ) {
				//
				// Do the query. It may not succeed, if we're waiting on other
				// values to be set (for example), so swallow the exception in
				// that case.
				//
				adobe::value_t result;
				try {
					db_m->query( query_m, boost::bind( &db_value_t::getResult, boost::ref( result ), _1 ) );
				} catch( ... ) {
					value_monitor_m( adobe::value_t() );
					if( active_monitor_m ) active_monitor_m( false );
					return;
				}
				value_monitor_m( result );
			}
			if( active_monitor_m ) active_monitor_m( !update_m.empty() );
		}
		void update_db( const adobe::value_t& db ) {
			std::string db_name = db.get<std::string>();
			if( db_name == db_name_m ) return;
			db_name_m = db_name;
			db_m = util::db::fromPath( scope_path_m / bfs::path( db_name_m, bfs::native ) );
			//
			// Make sure that we get notifications from the database.
			//
			db_connection_m = db_m->on_change( boost::bind( &db_value_t::update, this ) );
			update();
		}
		void update_query( const adobe::value_t& query ) {
			std::string query_str = query.get<std::string>();
			if( query_m == query_str ) return;
			query_m = query_str;
			update();
		}
		void update_update( const adobe::value_t& update ) {
			bool was_empty( update_m.empty() );
			if( update.get( update_m ) ) {
				if( (was_empty != update_m.empty()) && active_monitor_m ) active_monitor_m( !update_m.empty() );
			} else {
				if( active_monitor_m ) active_monitor_m( false );
			}
		}
	  public:
		db_value_t( const bfs::path& scope_path, const adobe::value_t& db, const adobe::value_t& query, const adobe::value_t& update, adobe::sheet_t& sheet )
		 : query_pump_m( sheet ), db_pump_m( sheet ), update_pump_m( sheet ), scope_path_m( scope_path ), sheet_m( sheet ), db_value_m( db ),
		   query_value_m( query ), update_value_m( update )
		{
			db_pump_m.initialize( db, boost::bind( &db_value_t::update_db, this, _1 ), 0 );
			query_pump_m.initialize( query, boost::bind( &db_value_t::update_query, this, _1 ), 0 );
			update_pump_m.initialize( update, boost::bind( &db_value_t::update_update, this, _1 ), 0 );
		}
		virtual ~db_value_t() {}
		virtual void set( const adobe::value_t& val ) {
			if( !db_m ) return;
			//
			// Walk through the update string, looking for "@" and ";".
			// 
			std::string update( update_m );
			std::string new_update( update_m );
			adobe::dictionary_t subst_dict;
			subst_dict.set( adobe::static_name_t( "NEWVAL" ), adobe::value_t( util::db::escape( val.get<std::string>() ) ) );
			do {
				update = new_update;
				new_update = substitute_dict( update, subst_dict );
			} while( update != new_update );
			try {
				db_m->execute( update );
			} catch( std::exception& ex ) {
				//
				// XXX: remove debug code.
				//
				std::cerr << "ERROR: " << ex.what() << " RUNNING: " << update << std::endl;
			}
		}
		virtual void set_signals( const adobe::sheet_t::monitor_value_t& monitor_value, const adobe::sheet_t::monitor_active_t& monitor_active ) {
			value_monitor_m = monitor_value;
			active_monitor_m = monitor_active;
			update();
		}
		virtual void substitute( const adobe::name_t& name, const adobe::value_t& val ) {
			db_pump_m.substitute( name, val );
			query_pump_m.substitute( name, val );
			update_pump_m.substitute( name, val );
		}
		virtual void invoke() {}
		virtual adobe::binder_ptr_t copy() const {
			return adobe::binder_ptr_t( new db_value_t( scope_path_m, db_value_m, query_value_m, update_value_m, sheet_m ) );
		}
	};
	adobe::value_t make_db_value_datapump( const bfs::path& scope_path, const adobe::dictionary_t& parameters, adobe::sheet_t& sheet ) {
		adobe::value_t db( parameters.at( adobe::static_name_t( "db" ) ) );
		adobe::value_t query( parameters.at( adobe::static_name_t( "query" ) ) );
		adobe::value_t update;
		if( parameters.count( adobe::static_name_t( "update" ) ) )
			update = parameters.at( adobe::static_name_t( "update" ) );
		return adobe::value_t( adobe::binder_ptr_t( new db_value_t( scope_path, db, query, update, sheet ) ) );
	}
}
